/*
 * 描          述:  <描述>
 * 修  改   人:  Administrator
 * 修改时间:  2014年6月19日
 * <修改描述:>
 */
package com.tx.component.dailytask.executor.impl;

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import com.tx.component.dailytask.executor.AbstractDailyTaskExecutor;
import com.tx.component.dailytask.task.impl.AsyncBatchDailyTask;
import com.tx.component.task.exeception.TaskExecuteException;
import com.tx.core.exceptions.util.AssertUtils;
import com.tx.core.util.MessageUtils;

/**
 * 线程池参数： https://blog.csdn.net/riemann_/article/details/104704197
 * 线程池关闭： https://blog.csdn.net/qq_39221436/article/details/123863354
 * 批量任务执行器<br/>
 *     这里这个实现是有问题的，需要进行调整<br/>
 * @author  Administrator
 * @version  [版本号, 2014年6月19日]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class AsyncBatchDailyTaskExecutor<T>
        extends AbstractDailyTaskExecutor<AsyncBatchDailyTask<T>>
        implements InitializingBean {
    
    /** 事务模板类 */
    private TransactionTemplate localTransTemplate;
    
    /** 任务执行器 */
    private ThreadPoolTaskExecutor taskExecutor;
    
    /** <默认构造函数> */
    public AsyncBatchDailyTaskExecutor() {
        super();
    }
    
    /** <默认构造函数> */
    public AsyncBatchDailyTaskExecutor(String taskBeanName,
            AsyncBatchDailyTask<T> task,
            PlatformTransactionManager transactionManager) {
        super(taskBeanName, task, transactionManager);
    }
    
    /**
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        AssertUtils.notEmpty(this.beanName, "beanName is empty.");
        AssertUtils.notNull(task, "task is null.");
        AssertUtils.notNull(transactionManager, "transactionManager is null.");
        
        this.localTransTemplate = new TransactionTemplate(
                this.transactionManager, new DefaultTransactionDefinition(
                        TransactionDefinition.PROPAGATION_REQUIRES_NEW));
        
        //线程池任务执行器
        this.taskExecutor = new ThreadPoolTaskExecutor();
        //核心线程数
        this.taskExecutor.setCorePoolSize(
                this.task.getPoolSize() <= 1 ? 1 : this.task.getPoolSize());
        //指定最大线程数
        this.taskExecutor.setMaxPoolSize(
                this.task.getPoolSize() <= 1 ? 1 : this.task.getPoolSize());
        //队列中最大的数目
        //this.taskExecutor.setQueueCapacity(650);
        //线程名称前缀
        this.taskExecutor
                .setThreadNamePrefix("DTP_TASK_" + this.task.getId() + "_:");
        //rejection-policy：当pool已经达到max size的时候，如何处理新任务
        //CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
        //对拒绝task的处理策略
        //this.taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //线程空闲后的最大存活时间
        //this.taskExecutor.setKeepAliveSeconds(60);
        this.taskExecutor.afterPropertiesSet();
    }
    
    /**
     * 多线程任务主要调度逻辑<br/>
     * @param date
     * @return
     */
    @Override
    public LocalDate doExecute(LocalDate date) {
        //获取任务总数，用以核对最后执行任务数量
        List<T> datas = this.task.list(date);
        int total = datas.size();
        final int batchSize = this.task.getBatchSize();
        final Queue<T> dataQueue = new ConcurrentLinkedQueue<>(datas);
        
        // 创建 Future 集合，用来存储Future 实例
        List<Future<Integer>> futureList = new ArrayList<>();
        while (!dataQueue.isEmpty()) {
            List<T> datasTemp = new ArrayList<>();
            for (int i = 0; i < batchSize && !dataQueue.isEmpty(); i++) {
                //移除并返问队列头部的元素
                T data = dataQueue.poll();
                datasTemp.add(data);
            }
            final AsyncBatchDailyTask<T> finalTask = this.task;
            /** 调用方法 */
            Future<Integer> future = this.taskExecutor
                    .submit(new Callable<Integer>() {
                        @Override
                        public Integer call() throws Exception {
                            return doExecuteTask(datasTemp, finalTask, date);
                        }
                    });
            //将结果压入堆栈
            futureList.add(future);
        }
        
        int count = 0;
        for (Future<Integer> future : futureList) {
            // 获取每一个线程的返回结果，如果该线程未完成，会阻塞
            try {
                count = count + future.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new TaskExecuteException(
                        MessageUtils.format("taskId:{}.任务执行错误.异常:{}",
                                this.task.getId(),
                                e.getMessage()),
                        e);
            }
        }
        //if (count != total) {
        //    throw new TaskExecuteException(
        //            MessageUtils.format("taskId:{}.任务执行错误.计数器:{};总任务数:{}",
        //                    this.task.getId(),
        //                    String.valueOf(count),
        //                    String.valueOf(total)));
        //}
        this.logger.info("taskId:{}.任务执行错误.计数器:{};总任务数:{}",
                new Object[] { this.task.getId(), String.valueOf(count),
                        String.valueOf(total) });
        //获取下次任务执行的时间
        LocalDate nextExecuteLocalDate = this.task.end(date);
        return nextExecuteLocalDate;
    }
    
    /**
     * 任务执行,返回执行的计数器<br/>
     * <功能详细描述>
     * @param datas
     * @param task
     * @param date
     * @return [参数说明]
     * 
     * @return int [返回类型说明]
     * @exception throws [异常类型] [异常说明]
     * @see [类、类#方法、类#成员]
     */
    private int doExecuteTask(final List<T> datas,
            final AsyncBatchDailyTask<T> task, final LocalDate date) {
        int count = this.localTransTemplate
                .execute(new TransactionCallback<Integer>() {
                    @Override
                    public Integer doInTransaction(TransactionStatus status) {
                        for (T data : datas) {
                            task.execute(data, date);
                        }
                        return datas.size();
                    }
                });
        return count;
    }
}
